package demo.mq.rocket.client;

import java.util.List;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;

public class ConcurrentlyMessageListenerDemo implements MessageListenerConcurrently {

    /**
     * 默认msgs里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        MessageExt messageExt = msgs.get(0);
        messageExt.getKeys();
        long offset = messageExt.getQueueOffset();
        String maxOffset = messageExt.getProperty(MessageConst.PROPERTY_MAX_OFFSET);
        long diff = Long.parseLong(maxOffset) - offset;
        /** 当某个队列的消息数堆积到100000条以上，则尝试丢弃部分或全部消息，这样就可以快速追上发送消息的速度。 */
        if (diff > 10_0000) {
            // TODO 消息堆积情况的特殊处理
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        // TODO 正常消费过程
        System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);

        if (messageExt.getTopic().equals("broker-a")) {
            // 执行TopicTest1的消费逻辑
            if (messageExt.getTags() != null && messageExt.getTags().equals("TagA")) {
                // 执行TagA的消费
                String message = new String(messageExt.getBody());
                System.out.println(message);
            } else if (messageExt.getTags() != null && messageExt.getTags().equals("TagB")) {
                // 执行TagB的消费
                String message = new String(messageExt.getBody());
                System.out.println(message);
            }

        } else if (messageExt.getTopic().equals("TopicTest2")) {
            // 执行TopicTest2的消费逻辑
            System.out.println("TopicTest2");
        }
        // 消费者向mq服务器返回消费成功的消息
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
